package ln.gather.jsr.jcache;

import ln.gather.jsr.jcache.event.CacheEntryEventPublisher;
import ln.gather.jsr.jcache.event.GanericCacheEntryEvent;
import ln.gather.jsr.jcache.integration.CompositeFallbackStorage;
import ln.gather.jsr.jcache.integration.FallbackStorage;
import ln.gather.jsr.jcache.management.CacheStatistics;
import ln.gather.jsr.jcache.management.DummyCacheStatistics;
import ln.gather.jsr.jcache.management.ManagementUtils;
import ln.gather.jsr.jcache.management.SimpleCacheStatistics;
import ln.gather.jsr.jcache.configuration.ConfigurationUtils;
import ln.gather.jsr.jcache.processor.MutableEntryAdaptor;

import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.CacheManager;
import javax.cache.configuration.*;
import javax.cache.event.CacheEntryListener;
import javax.cache.expiry.Duration;
import javax.cache.expiry.EternalExpiryPolicy;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.integration.CacheLoader;
import javax.cache.integration.CacheWriter;
import javax.cache.integration.CompletionListener;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;
import javax.cache.processor.MutableEntry;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Supplier;
import java.util.logging.Logger;

/**
 * @Author zhangj
 * @Date 2022/11/17 21:38
 * @see Cache
 */
public abstract class AbstractCache<K, V> implements Cache<K, V> {

    protected final Logger logger = Logger.getLogger(this.getClass().getName());

    private final CacheManager cacheManager;

    private final String cacheName;

    private final MutableConfiguration<K, V> configuration;

    private final ExpiryPolicy expiryPolicy;

    private final CacheLoader<K, V> cacheLoader;

    private final CacheWriter<K, V> cacheWriter;

    private final FallbackStorage defaultFallbackStorage;

    private final CacheEntryEventPublisher entryEventPublisher;

    private final CacheStatistics cacheStatistics;

    private final Executor executor;

    private volatile boolean closed = false;

    public AbstractCache(CacheManager cacheManager, String cacheName, Configuration<K, V> configuration) {
        this.cacheManager = cacheManager;
        this.cacheName = cacheName;
        this.configuration = ConfigurationUtils.mutableConfiguration(configuration);
        this.expiryPolicy = resolveExpiryPolicy(getConfiguration());
        this.defaultFallbackStorage = new CompositeFallbackStorage();
        this.cacheLoader = resolveCacheLoader(getConfiguration());
        this.cacheWriter = resolveCacheWriter(getConfiguration());
        this.entryEventPublisher = new CacheEntryEventPublisher();
        this.cacheStatistics = resolveCacheStatistics();
        this.executor = ForkJoinPool.commonPool();
        registerCacheEntryListenersFromConfiguration();
        registerMBeansIfRequired(this, this.cacheStatistics);
    }

    private void registerMBeansIfRequired(Cache<K, V> cache, CacheStatistics cacheStatisics) {
        CompleteConfiguration configuration = cache.getConfiguration(CompleteConfiguration.class);
        if (configuration.isStatisticsEnabled()) {
            MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
            registerCacheMXBeanIfRequire(cache, configuration, mBeanServer);
            registerCacheStatisticsMXBeanIfRequired(cache, configuration, mBeanServer, cacheStatisics);
        }
    }

    private static <K, V> void registerCacheStatisticsMXBeanIfRequired(Cache<?, ?> cache, CompleteConfiguration configuration,
                                                                       MBeanServer mBeanServer, CacheStatistics cacheStatisics) {
        if (configuration.isStatisticsEnabled()) {
            ObjectName objectName = ManagementUtils.createObjectName(cache, "CacheStatistics");
            registerMBean(objectName, cacheStatisics, mBeanServer);
        }
    }

    private static <K, V> void registerCacheMXBeanIfRequire(Cache<K, V> cache, CompleteConfiguration configuration, MBeanServer mBeanServer) {
        ObjectName objectName = ManagementUtils.createObjectName(cache, "CacheConfiguration");
        registerMBean(objectName, ManagementUtils.adaptCacheMXBean(configuration), mBeanServer);
    }

    private static void registerMBean(ObjectName objectName, Object object, MBeanServer mBeanServer) {
        if (!mBeanServer.isRegistered(objectName)) {
            try {
                mBeanServer.registerMBean(object, objectName);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

    }
    private void registerCacheEntryListenersFromConfiguration() {
        this.getConfiguration().getCacheEntryListenerConfigurations().forEach(this::registerCacheEntryListener);
    }

    @Override
    public void deregisterCacheEntryListener(CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration) {
        entryEventPublisher.deregisterCacheEntryListener(cacheEntryListenerConfiguration);
    }

    protected final boolean isStatisticsEnable() {
        return configuration.isStatisticsEnabled();
    }

    protected final boolean isReadThrough() {
        return configuration.isReadThrough();
    }

    protected final boolean isWriteThrough() {
        return configuration.isWriteThrough();
    }

    public CacheLoader<K, V> getCacheLoader() {
        return cacheLoader;
    }

    public CacheWriter<K, V> getCacheWriter() {
        return cacheWriter;
    }

    @Override
    public String getName() {
        return cacheName;
    }

    @Override
    public CacheManager getCacheManager() {
        return cacheManager;
    }

    @Override
    public boolean isClosed() {
        return closed;
    }

    @Override
    public <T> T unwrap(Class<T> clazz) {
        return getCacheManager().unwrap(clazz);
    }

    @Override
    public final void close() {
        if(isClosed()) {
            return;
        }
        doClose();
        closed = true;
    }

    /**
     * Subclass could override this method.
     */
    protected void doClose() {
    }

    @Override
    public Iterator<Entry<K, V>> iterator() {
        assertNotClosed();
        List<Entry<K, V>> entries = new LinkedList<>();
        keySet().forEach(key -> {
            entries.add(ExpirableEntry.of(key, get(key)));
        });
        return entries.iterator();
    }

    private CacheStatistics resolveCacheStatistics() {
        return isStatisticsEnable() ?
                new SimpleCacheStatistics() : DummyCacheStatistics.INSTANCE;
    }


    private ExpiryPolicy resolveExpiryPolicy(CompleteConfiguration<?, ?> configuration) {
        Factory<ExpiryPolicy> expiryPolicyFactory = configuration.getExpiryPolicyFactory();
        if (expiryPolicyFactory == null) {
            expiryPolicyFactory = EternalExpiryPolicy::new;
        }
        return expiryPolicyFactory.create();
    }

    protected final CompleteConfiguration<K, V> getConfiguration() {
        return this.configuration;
    }

    private CacheLoader<K, V> resolveCacheLoader(CompleteConfiguration<K, V> configuration) {
        Factory<CacheLoader<K, V>> cacheLoaderFactory = configuration.getCacheLoaderFactory();
        CacheLoader<K, V> cacheLoader = null;
        if (cacheLoaderFactory != null) {
            cacheLoader = cacheLoaderFactory.create();
        }

        if (cacheLoader == null) {
            cacheLoader = defaultFallbackStorage;
        }
        return cacheLoader;
    }

    private CacheWriter<K, V> resolveCacheWriter(CompleteConfiguration<K, V> configuration) {
        Factory<CacheWriter<? super K, ? super V>> cacheWriterFactory = configuration.getCacheWriterFactory();
        CacheWriter<K, V> cacheWriter = null;
        if (cacheWriterFactory != null) {
            cacheWriter = (CacheWriter<K, V>) cacheWriterFactory.create();
        }

        if (cacheWriter == null) {
            cacheWriter = defaultFallbackStorage;
        }
        return cacheWriter;
    }

    /**
     * 发布缓存创建消息
     */
    private void publishCreatedEvent(K key, V value) {
        entryEventPublisher.publish(GanericCacheEntryEvent.createEvent(this, key, value));
    }

    /**
     * 发布缓存修改消息
     */
    private void publishUpdatedEvent(K key, V oldValue, V value) {
        entryEventPublisher.publish(GanericCacheEntryEvent.updateEvent(this, key, oldValue, value));
    }

    /**
     * 发布缓存过期消息
     */
    private void publishExpiredEvent(K key, V value) {
        entryEventPublisher.publish(GanericCacheEntryEvent.expiredEvent(this, key, value));
    }

    /**
     * 发布缓存清除消息
     */
    private void publishRemovedEvent(K key, V value) {
        entryEventPublisher.publish(GanericCacheEntryEvent.removeEvent(this, key, value));
    }

    private Duration getDurationForCreation() {
        return getDuration(expiryPolicy::getExpiryForCreation);
    }

    private Duration getDurationForAccess() {
        return getDuration(expiryPolicy::getExpiryForAccess);
    }

    private Duration getDurationForUpdate() {
        return getDuration(expiryPolicy::getExpiryForUpdate);
    }

    private Duration getDuration(Supplier<Duration> durationSupplier) {
        Duration duration = null;
        try {
            duration = durationSupplier.get();
        } catch (Throwable e) {
            duration = Duration.ETERNAL;
        }
        return duration;
    }

    private boolean handleExpiryPolicyForCreation(ExpirableEntry<K, V> newEntry) {
        return handleExpiryPolicy(newEntry, getDurationForCreation(), false);
    }

    private boolean handleExpiryPolicyForAccess(ExpirableEntry<K, V> newEntry) {
        return handleExpiryPolicy(newEntry, getDurationForAccess(), true);
    }

    private boolean handleExpiryPolicyForUpdate(ExpirableEntry<K, V> newEntry) {
        return handleExpiryPolicy(newEntry, getDurationForUpdate(), true);
    }

    /**
     * 过期策略
     */
    private boolean handleExpiryPolicy(ExpirableEntry<K, V> entry, Duration duration, boolean removeExpireEntry) {
        if (entry == null) {
            return false;
        }
        boolean expire = false;
        if (entry.isExpired()) {
            expire = true;
        } else if (duration != null) {
            if (duration.isZero()) {
                expire = true;
            } else {
                long timestamp = duration.getAdjustedTime(System.currentTimeMillis());
                entry.setTimestamp(timestamp);
            }
        }

        if (removeExpireEntry && expire) {
            K key = entry.getKey();
            V value = entry.getValue();
            removeEntry(key);
            publishExpiredEvent(key, value);
            cacheStatistics.cacheEvictions();
        }

        return expire;
    }

    private V loadValue(K key) {
        return getCacheLoader().load(key);
    }

    private V loadValue(K key, boolean storedEntry) {
        V value = loadValue(key);
        if (storedEntry && value != null) {
            put(key, value);
        }
        return value;
    }

    private void writeEntryIfWriteThrough(Entry<K, V> entry) {
        if (entry != null && isWriteThrough()) {
            getCacheWriter().write(entry);
        }
    }

    private void deleteEntryIfWriteThrough(K key) {
        if (key != null && isWriteThrough()) {
            getCacheWriter().delete(key);
        }
    }

    private ExpirableEntry<K, V> createEntry(K key, V value) {
        return ExpirableEntry.of(key, value);
    }

    private Entry<K, V> createAndPutEntry(K key, V value) {
        ExpirableEntry newEntry = createEntry(key, value);
        if (handleExpiryPolicyForCreation(newEntry)) {
            // The new Cache.Entry is already expired and will not be added to the cache.
            return null;
        }

        putEntry(newEntry);
        publishCreatedEvent(key, value);
        return newEntry;
    }

    private Entry<K, V> updateEntry(K key, V value) {
        ExpirableEntry<K, V> oldEntry = getEntry(key);
        V oldValue = oldEntry.getValue();
        oldEntry.setValue(value);
        putEntry(oldEntry);
        publishUpdatedEvent(key, oldValue, value);
        if (handleExpiryPolicyForUpdate(oldEntry)) {
            return null;
        }
        return oldEntry;
    }

    private static <K, V> V getValue(Entry<K, V> entry) {
        return entry == null ? null : entry.getValue();
    }

    @Override
    public V get(K key) {
        assertNotClosed();
        Objects.requireNonNull(key);
        ExpirableEntry<K, V> entry = null;
        V value = null;
        long startTime = System.currentTimeMillis();
        try {
            entry = getEntry(key);
            if (handleExpiryPolicyForAccess(entry)) {
                return null;
            }
            if (entry == null && isReadThrough()) {
                // 缓存穿透时补偿查询
                value = loadValue(key, true);
            } else {
                value = getValue(entry);
            }
        } catch (Throwable e) {
            logger.severe(e.getMessage());
        } finally {
            if (value != null) {
                cacheStatistics.cacheHits();
            }
            cacheStatistics.cacheGets();
            cacheStatistics.cacheGetsTime(System.currentTimeMillis() - startTime);
        }
        return value;
    }

    @Override
    public Map<K, V> getAll(Set<? extends K> keys) {
        Map<K, V> reMap = new LinkedHashMap<>();
        keys.forEach(key -> {
            reMap.put(key, get(key));
        });
        return reMap;
    }

    @Override
    public boolean containsKey(K key) {
        assertNotClosed();
        return containsEntry(key);
    }

    /**
     * {@inheritDoc}
     *
     * @param keys                  the keys to load
     * @param replaceExistingValues when true existing values in the Cache will
     *                              be replaced by those loaded from a CacheLoader
     * @param completionListener    the CompletionListener (may be null)
     */
    @Override
    public void loadAll(Set<? extends K> keys, boolean replaceExistingValues, CompletionListener completionListener) {
        assertNotClosed();
        if (!isReadThrough()) {
            // FIXME: The specification does not mention that
            completionListener.onCompletion();
            return;
        }

        CompletableFuture.runAsync(() -> {
            for (K key : keys) {
                V value = loadValue(key, true);
                if (replaceExistingValues) {
                    replace(key, value);
                } else {
                    put(key, value);
                }
            }
        }, executor).whenComplete((v, e) -> {
            if (completionListener != null) {
                if (e instanceof Exception && e.getCause() instanceof Exception) {
                    completionListener.onException((Exception) e.getCause());
                } else {
                    completionListener.onCompletion();
                }
            }
        });
    }

    /**
     * {@inheritDoc}
     * <p>
     * Current method calls the methods of {@link ExpiryPolicy}:
     * <ul>
     *     <li>Yes {@link ExpiryPolicy#getExpiryForCreation} (when the key is not associated with an existing value)</li>
     *     <li>No {@link ExpiryPolicy#getExpiryForAccess}</li>
     *     <li>Yes {@link ExpiryPolicy#getExpiryForUpdate} (when the key is associated with an existing value)</li>
     * </ul>
     *
     * @param key   key with which the specified value is to be associated
     * @param value value to be associated with the specified key
     */
    @Override
    public void put(K key, V value) {
        assertNotClosed();
        Entry<K, V> entry = null;
        long startTime = System.currentTimeMillis();
        try {
            if (!containsKey(key)) {
                entry = createAndPutEntry(key, value);
            } else {
                entry = updateEntry(key, value);
            }
        } finally {
            writeEntryIfWriteThrough(entry);
            cacheStatistics.cachePuts();
            cacheStatistics.cachePutsTime(System.currentTimeMillis() - startTime);
        }
    }

    /**
     * {@inheritDoc}
     *
     * @param key   key with which the specified value is to be associated
     * @param value value to be associated with the specified key
     * @return
     */
    @Override
    public V getAndPut(K key, V value) {
        Entry<K, V> oldEntry = getEntry(key);
        V oldValue = oldEntry.getValue();
        put(key, value);
        return oldValue;
    }

    /**
     * {@inheritDoc}
     *
     * @param map mappings to be stored in this cache
     */
    @Override
    public void putAll(Map<? extends K, ? extends V> map) {
        map.forEach((k, v) -> put(k, v));
    }

    /**
     * {@inheritDoc}
     *
     * @param key   key with which the specified value is to be associated
     * @param value value to be associated with the specified key
     * @return
     */
    @Override
    public boolean putIfAbsent(K key, V value) {
        if (!containsEntry(key)) {
            put(key, value);
            return true;
        } else {
            return false;
        }
    }

    /**
     * {@inheritDoc}
     * @param key key whose mapping is to be removed from the cache
     * @return
     */
    @Override
    public boolean remove(K key) {
        assertNotClosed();
        Objects.requireNonNull(key);
        boolean removed = false;
        long startTime = System.currentTimeMillis();
        try {
            ExpirableEntry<K, V> oldEntry = removeEntry(key);
            removed = oldEntry != null;
            if(removed) {
                publishRemovedEvent(key, oldEntry.getValue());
            }
        } finally {
            deleteEntryIfWriteThrough(key);
            cacheStatistics.cacheRemovals();
            cacheStatistics.cacheRemovesTime(System.currentTimeMillis() - startTime);
        }
        return removed;
    }

    /**
     * {@inheritDoc}
     * @param key      key whose mapping is to be removed from the cache
     * @param oldValue value expected to be associated with the specified key
     * @return
     */
    @Override
    public boolean remove(K key, V oldValue) {
        if(containsEntry(key) && Objects.equals(oldValue, get(key))) {
            return remove(key);
        } else {
            return false;
        }
    }

    @Override
    public V getAndRemove(K key) {
        Entry<K, V> oldEntry = getEntry(key);
        V oldValue = oldEntry.getValue();
        remove(key);
        return oldValue;
    }

    @Override
    public boolean replace(K key, V value) {
        if(containsEntry(key)) {
            put(key, value);
            return true;
        } else {
            return false;
        }
    }

    @Override
    public boolean replace(K key, V oldValue, V newValue) {
        Objects.requireNonNull(oldValue);
        if(containsKey(key) && Objects.equals(get(key), oldValue)) {
            put(key, newValue);
            return true;
        } else {
            return false;
        }
    }

    @Override
    public V getAndReplace(K key, V value) {
        Entry<K, V> oldEntry = getEntry(key);
        V oldValue = oldEntry.getValue();
        put(key, value);
        return oldValue;
    }

    @Override
    public void removeAll(Set<? extends K> keys) {
        keys.forEach(this::remove);
    }

    @Override
    public void removeAll() {
        removeAll(keySet());
    }

    @Override
    public void clear() {
        assertNotClosed();
        clearEntries();
        defaultFallbackStorage.destroy();
        cacheStatistics.reset();
    }

    @Override
    public <C extends Configuration<K, V>> C getConfiguration(Class<C> clazz) {
        if(!Configuration.class.isAssignableFrom(clazz)) {
            throw new IllegalArgumentException("必须继承自" + Configuration.class.getName());
        }
        return (C) ConfigurationUtils.immutableConfiguration(getConfiguration());
    }

    /**
     * 通过 key 将对应的数据进行转换
     *
     * @param key
     * @param entryProcessor
     * @param arguments
     * @param <T>
     * @return
     * @throws EntryProcessorException
     */
    @Override
    public <T> T invoke(K key, EntryProcessor<K, V, T> entryProcessor, Object... arguments) throws EntryProcessorException {
        MutableEntry<K, V> mutableEntry = MutableEntryAdaptor.of(key, this);
        return entryProcessor.process(mutableEntry, arguments);
    }

    /**
     * 通过多个 key 将对应的数据进行转换
     *
     * @param keys
     * @param entryProcessor 进行转换，将Entry转换为其他的东西
     * @param arguments
     * @param <T>
     * @return
     */
    @Override
    public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys, EntryProcessor<K, V, T> entryProcessor, Object... arguments) {
        Map<K, EntryProcessorResult<T>> reMap = new LinkedHashMap<>();
        keys.forEach(key -> {
            reMap.put(key, () -> invoke(key, entryProcessor, arguments));
        });
        return reMap;
    }

    /**
     * 注册监听器<br/>
     *
     * @param cacheEntryListenerConfiguration
     * @see CacheEntryListener 四大监听器
     */
    @Override
    public void registerCacheEntryListener(CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration) {

    }

    private void assertNotClosed() {
        if (isClosed()) {
            throw new IllegalStateException("Current cache has been closed! No operation should be executed.");
        }
    }

    /**
     * Get all keys of {@link Cache.Entry} in the {@link Cache}
     *
     * @return the non-null read-only {@link Set}
     */
    protected abstract Set<K> keySet();

    /**
     * Clear all {@link Cache.Entry enties} from cache.
     *
     * @throws CacheException if there is a problem doing the clear
     */
    protected abstract void clearEntries() throws CacheException;
    
    protected abstract ExpirableEntry<K, V> removeEntry(K key) throws CacheException, ClassCastException;

    protected abstract ExpirableEntry<K, V> getEntry(K key) throws CacheException, ClassCastException;

    protected abstract boolean containsEntry(K key) throws CacheException, ClassCastException;

    protected abstract void putEntry(ExpirableEntry<K, V> entry) throws CacheException, ClassCastException;

}
